package com.atguigu.dw.gamallcanal


import java.util
import java.util.Random

import com.alibaba.fastjson.JSONObject
import com.alibaba.otter.canal.protocol.CanalEntry
import com.alibaba.otter.canal.protocol.CanalEntry.{EventType, RowChange, RowData}
import com.atguigu.constan.ConstanVal
import com.google.protobuf.ByteString

import scala.collection.JavaConversions._

/**
 * Author lzc
 * Date 2019/5/17 4:09 PM
 */
object CanalHandler {

  /**
   * 处理监听到变化的内容
   *
   * @Author 剧情再美终是戏
   * @Date 2020/2/14 14:58
   * @param entries
   * @return java.lang.String
   * @Version 1.0
   **/
  def handle(entries: util.List[CanalEntry.Entry]) = {
    // 一个 en 就是一条sql
    for (en <- entries) {
      if (null != en) {
        val header = en.getHeader
        if (ConstanVal.MYSQL_TABLE_ORDER_INFO.equals(header.getTableName) // order_info 表 insert 事件
          && header.getEventType == EventType.INSERT) {
          handleDataToKafka(ConstanVal.KAFKA_TOPIC_ORDER, en)
        }
        if (ConstanVal.MYSQL_TABLE_ORDER_DETAIL.equals(header.getTableName) // order_detail 表 insert 事件
          && header.getEventType == EventType.INSERT) {
          handleDataToKafka(ConstanVal.KAFKA_TOPIC_ORDER_DETAIL, en)
        }
      }
    }
  }


  /**
   * 处理 entry 数据并入库 kafka
   *
   * @Author 剧情再美终是戏
   * @Date 2020/2/18 20:46
   * @param topic 入库kafka 主题
   * @param en    canal 监听到变化的Entry
   * @return void
   * @Version 1.0
   **/
  private def handleDataToKafka(topic: String, en: CanalEntry.Entry) = {
    new Thread {
      override def run() = {
//        Thread.sleep(new Random().nextInt(10) * 1000)
        // getStoreValue 这是一条sql，变化的所有内容
        val value: ByteString = en.getStoreValue

        // 获取变化内容所影响的行
        val change = RowChange.parseFrom(value)
        val list = change.getRowDatasList

        // 处理第一行变化的数据
        for (li <- list) {
          val jsonObj = new JSONObject()
          // 处理第一列变化的数据
          val columnList = li.getAfterColumnsList
          for (cll <- columnList) {
            jsonObj.put(cll.getName, cll.getValue)
          }
          // 入库 kafka
          MyKafkaUtil.send(topic, jsonObj.toJSONString)
          println(s"入库kafka 成功 topic:$topic value : " + jsonObj.toJSONString)
        }
      }
    }.start()
  }


  /**
   * 处理从 canal 取来的数据
   *
   * @param tableName   表名
   * @param eventType   事件类型
   * @param rowDataList 数据类别
   */
  def handle(tableName: String, eventType: EventType, rowDataList: util.List[RowData]) = {
    import scala.collection.JavaConversions._
    if ("order_info" == tableName && eventType == EventType.INSERT && rowDataList.size() > 0) {
      // 1. rowData 表示一行数据, 通过他得到每一列. 首先遍历每一行数据
      for (rowData <- rowDataList) {
        // 2. 得到每行中, 所有列组成的列表
        val columnList: util.List[CanalEntry.Column] = rowData.getAfterColumnsList
        for (column <- columnList) {
          // 3. 得到列名和列值
          println(column.getName + ":" + column.getValue)
        }
      }
    }
  }
}